package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.util.DimUtil;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        String sinkTable = value.getString("sink_table");
        value.remove("sink_table");

        String type = value.getString("type");
      value.remove("type");

        String upsertSql = "upsert into " + EduConfig.PHOENIX_SCHEMA + "." + sinkTable
                + "(" + StringUtils.join(value.keySet(), ",") + ") " +
                " values('" + StringUtils.join(value.values(), "','") + "')";
        System.out.println("向phoenix表中插入数据的SQL:" + upsertSql);

        PhoenixUtil.executeSql(upsertSql);
        if ("update".equals(type)){
            DimUtil.deleteCached(sinkTable,value.getString("id"));
        }

    }
}
